package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.EduConfig;
import com.atguigu.util.DruidPhoenixDSUtil;
import com.atguigu.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.util.*;

/**
 * @author hfh
 * @version 1.0
 * @date 2022/8/15 11:35
 */
public class MyBroadcastFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private DruidDataSource druidDataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource=DruidPhoenixDSUtil.getDataSource();
    }

    public MyBroadcastFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    /**
     * 主流
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 1 获取状态  判断当前表是否为维度表  不是删除
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String table = value.getString("table");
        TableProcess tableConfig = broadcastState.get(table);
        if (tableConfig!=null){
            // 2 过滤掉多余的字段  只保留sink_columns字段
            JSONObject data = value.getJSONObject("data");
//            System.out.println("过滤字段前："+data.toJSONString());
            String sinkTable = tableConfig.getSinkTable();
//            filterColumns(data,tableConfig.getSinkColumns());

            // 3 添加sink_table字段
            data.put("sink_table",sinkTable);
//            System.out.println("过滤字段后："+data.toJSONString());
            out.collect(data);

        }


    }

    private void filterColumns(JSONObject data, String sinkColumns) {
        //保留下sinkColumns中的字段，没有的删除
        Set<Map.Entry<String, Object>> dataEntries = data.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = dataEntries.iterator();
        String[] split = sinkColumns.split(",");
        List<String> cols = Arrays.asList(split);
        while(iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            if (!cols.contains(next.getKey())){
                iterator.remove();
            }
        }

    }
    /**
     * 广播流处理
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        JSONObject jsonObject = JSONObject.parseObject(value);
        // 1 判断类型是否是d  如果是删除  从状态中去除对应的表
        String op = jsonObject.getString("op");
        if ("d".equals(op)){
            TableProcess before = jsonObject.getObject("before", TableProcess.class);
            broadcastState.remove(before.getSourceTable());
        }else{
            TableProcess after = jsonObject.getObject("after", TableProcess.class);
            String sourceTable = after.getSourceTable();
            String sinkTable = after.getSinkTable();
            String sinkColumns = after.getSinkColumns();
            String sinkPk = after.getSinkPk();
            String sinkExtend = after.getSinkExtend();
            // 2 判断表格是否在phoenix中存在  不存在创建
            checkTable(sourceTable,sinkTable,sinkColumns,sinkPk,sinkExtend);
            // 3 将内容写入状态
            broadcastState.put(after.getSourceTable(),after);
        }
    }
    //此方法创建表格不支持复合主键
    private void checkTable(String sourceTable, String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        //2 判断主键和扩展信息是否为空，是则给默认值
        if(sinkPk==null){
            sinkPk="id";

        }

        if (sinkExtend==null){
            sinkExtend="";
        }
        //1.拼接一个sql语句
        StringBuilder sql = new StringBuilder();
        sql.append("create table if not exists ")
                .append(EduConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append(" (");

        String[] cols = sinkColumns.split(",");
        for (int i = 0; i < cols.length; i++) {
            //判断是否为主键
            if (cols[i].equals(sinkPk)){
                sql.append(cols[i])
                        .append(" varchar primary key ");
            }else{
                sql.append(cols[i])
                        .append(" varchar ");
            }

            if (i<cols.length-1){
                sql.append(",");
            }
        }

        sql.append(")").append(sinkExtend);
        String createSql = sql.toString();
        System.out.println(createSql);

        //连接phoenix执行建表语句
        DruidPooledConnection connection=null;
        try {
            connection = druidDataSource.getConnection();
            PhoenixUtil.executeSql(createSql,connection);
            System.out.println("维度表建表语句执行完毕！");
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

}
